package com.bit.meta;

import com.bit.Constant;
import com.bit.util.NetUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * bit-supper-MetaService
 *
 * @author <a href="2428922347@qq.com">Zhu mingming</a>
 * @info
 * @since 2018/10/12 下午 01:38
 */
@Slf4j
@Service
public class MetaService implements InitializingBean {

    @Autowired
    private ZkUtil zkUtil;

    @Value("#{'${data.server.topic}'.split(',')}")
    private List<String> topics;

    private Map<String, PathWithNum> leadApplyInfo = new HashMap<>();

    @Override
    public void afterPropertiesSet() throws Exception {
        //创建根节点
        zkUtil.createPath("/" + Constant.ZK_ROOT_NODE, CreateMode.PERSISTENT, "");

        //创建topic节点
        for (String topic : topics) {
            registerNode(topic);
        }



    }


    protected void registerNode(String topic) throws Exception {

        String topicPath = "/" + Constant.ZK_ROOT_NODE + "/" + topic;

        //创建当前topic节点
        zkUtil.createPath(topicPath, CreateMode.PERSISTENT, "");

        //创建当前topic的工作节点
        zkUtil.createPath(topicPath + "/worker", CreateMode.PERSISTENT, "");
        zkUtil.createPath(topicPath + "/worker/" + NetUtil.getLocalIp(), CreateMode.EPHEMERAL, "");

        //创建当前topic的主节点
        zkUtil.createPath(topicPath + "/main", CreateMode.PERSISTENT, "");

        //申请成为主节点
        applyLeadShip(topic);

        //配置当前节点信息
        configEnvironment(topic);

    }

    private void configEnvironment(String topic) throws Exception {
        PathWithNum topicLeaderInfo = getTopicLeaderInfo(topic);
        if (topicLeaderInfo.getNum().equals(leadApplyInfo.get(topic).getNum())) {
            log.info("当前节点【{}】-->主节点！开始配置主节点。。。",topic);
        } else {
            log.info("当前节点-->备节点！开始监听。。。。");
            zkUtil.watch(topicLeaderInfo.getPath(), new Watcher() {
                @Override
                public void process(WatchedEvent watchedEvent) {
                    try {
                        configEnvironment(topic);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            });
        }
    }

    /**
     * 争取lead位置
     *
     * @param topic
     * @return
     * @throws Exception
     */
    protected void applyLeadShip(String topic) throws Exception {
        String rp = "/" + Constant.ZK_ROOT_NODE + "/" + topic + "/main";
        String path = zkUtil.createPath(rp + "/0", CreateMode.EPHEMERAL_SEQUENTIAL, NetUtil.getLocalIp());
        Integer selfFightPosition = Integer.parseInt(path.replace(rp + "/0", ""));
        leadApplyInfo.put(topic, new PathWithNum(path, selfFightPosition));
    }


    /**
     * 获取topic的主节点信息
     *
     * @param topic
     * @return
     * @throws Exception
     */
    public PathWithNum getTopicLeaderInfo(String topic) throws Exception {
        String rp = "/" + Constant.ZK_ROOT_NODE + "/" + topic + "/main";
        List<String> child = zkUtil.child(rp);
        List<PathWithNum> competitor =
                child.stream()
                        .map(s -> {
                            return new PathWithNum(rp + "/" + s, Integer.parseInt(s));
                        })
                        .collect(Collectors.toList());
        if (competitor == null || competitor.size() <= 0)
            throw new IllegalStateException("zookeeper数据异常，competitor数量应该大于等于1，当前为0");
        return getMin(competitor);
    }


    /**
     * 获取最早建立的节点
     *
     * @param competitor
     * @return
     */
    private PathWithNum getMin(List<PathWithNum> competitor) {
        PathWithNum min = null;
        for (PathWithNum pathWithNum : competitor) {
            PathWithNum withNum = (min == null ? (min = pathWithNum) : ((min.getNum() > pathWithNum.getNum()) ? (min = pathWithNum) : (min = min)));
        }
        return min;
    }

    /**
     * 获取工作节点
     *
     * @param topic
     * @return
     * @throws Exception
     */
    public List<String> getTopicWorker(String topic) throws Exception {
        String rp = "/" + Constant.ZK_ROOT_NODE + "/" + topic + "/worker";
        return zkUtil.child(rp);
    }

}
